package main

import (
	"encoding/json"
	"flag"
	"github.com/eclipse/paho.mqtt.golang"
	"github.com/fhmq/hmq/broker"
	"github.com/hashicorp/go-uuid"
	"github.com/valyala/fasthttp"
	"log"
	"os"
	"os/signal"
	"runtime"
	"strconv"
	"time"
)

var (
	addr            = flag.String("address", "0.0.0.0:8080", "http服务器地址")
	compress        = flag.Bool("compress", false, "是否启用压缩")
	mqttClient      = flag.String("mqtt_client", "tcp://127.0.0.1:1885", "mqtt 客户端要订阅的broker地址")
	mqttBrokerHost  = flag.String("mqtt_broker", "127.0.0.1", "mqtt 服务器要订阅的broker地址")
	mqttBrokerPort  = flag.String("mqtt_broker_port", "1885", "mqtt 服务器要订阅的broker地址")
	mqttPrefix      = flag.String("mqtt_topic_prefix", "", "http 转mqtt的 topic 前缀")
	httpHost        = flag.String("http_host", "http://127.0.0.1", "mqtt 转发给http的接口")
	callbackTimeout = flag.Int("callback_timeout", 2, "mqtt 返回数据的超时时间")
	httpPrefix      = flag.String("http_prefix", "", "mqtt 转http的 urlPath 前缀")
	c               mqtt.Client
)

func main() {
	flag.Parse()
	go startHttp()
	go startMqtt()
	s := waitForSignal()
	log.Println("关闭", s)
}

func startMqttClient() {
	log.Println("启动mqttclient")
	ops := mqtt.NewClientOptions().SetClientID("http2mqtt").AddBroker(*mqttClient).SetAutoReconnect(true)
	c = mqtt.NewClient(ops)

	if token := c.Connect(); token.Wait() && token.Error() != nil {
		log.Println(token.Error())
	}

	//log.Println("启动mqtt 转发 http")
	//c.Subscribe("/", 0, func(Client mqtt.Client, Message mqtt.Message) {
	//	log.Println("转发mqtt发起http请求" + *httpHost + " " + *httpPrefix)
	//	//发起http请求
	//})

}
func startMqtt() {
	log.Println("启动mqtt服务器 " + *mqttBrokerHost + ":" + *mqttBrokerPort)
	runtime.GOMAXPROCS(runtime.NumCPU())
	config := broker.Config{}
	config.Host = *mqttBrokerHost
	config.Port = *mqttBrokerPort

	b, err := broker.NewBroker(&config)
	if err != nil {
		log.Fatal("New Broker error: ", err)
	}
	b.Start()
	startMqttClient()

}
func startHttp() {
	log.Println("启动http服务器 " + *addr)
	h := requestHandler
	if *compress {
		h = fasthttp.CompressHandler(h)
	}

	if err := fasthttp.ListenAndServe(*addr, h); err != nil {
		log.Fatalf("Error in ListenAndServe: %s", err)
	}
}

func requestHandler(ctx *fasthttp.RequestCtx) {
	resId, _ := uuid.GenerateUUID()
	res := map[string]interface{}{
		"resId":  resId,
		"method": string(ctx.Request.Header.Method()),
		"body":   string(ctx.Request.Body()),
	}
	resRaw, _ := json.Marshal(res)
	log.Println("topic: " + string(ctx.Request.URI().FullURI()))
	log.Println("content: " + string(resRaw))

	repData := []byte("{\"msgtype\": \"empty\"}")
	var ch = make(chan bool)
	c.Subscribe("/"+resId, 1, func(Client mqtt.Client, Message mqtt.Message) {
		repData = Message.Payload()
		log.Println("收到请求 " + string(repData))
		ch <- true
	})
	c.Publish(string(ctx.Request.URI().FullURI()), 1, false, resRaw)
	log.Println("超时时间" + strconv.Itoa(*callbackTimeout))
	select {
	case <-time.After(time.Duration(*callbackTimeout*1000_000_000)):
		log.Println("默认内容" + string(repData))
	case <-ch:
		log.Println("回复" + string(repData))
	}

	_, _ = ctx.Write(repData)
}

func waitForSignal() os.Signal {
	signalChan := make(chan os.Signal, 1)
	defer close(signalChan)
	signal.Notify(signalChan, os.Kill, os.Interrupt)
	s := <-signalChan
	signal.Stop(signalChan)
	return s
}
